package com.itc.bi.handler.dataSet;

import cn.hutool.core.date.DatePattern;
import cn.hutool.core.date.DateUtil;
import com.alibaba.fastjson2.JSONArray;
import com.alibaba.fastjson2.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.core.toolkit.IdWorker;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.itc.bi.constant.DataDirectorConstant;
import com.itc.bi.constant.KafkaConstant;
import com.itc.bi.constant.SystemConstant;
import com.itc.bi.dto.DataSetCreateDto;
import com.itc.bi.entity.*;
import com.itc.bi.handler.DatasourceHandler;
import com.itc.bi.mapper.DataSynrecordMapper;
import com.itc.bi.mapper.DirectoryMapper;
import com.itc.bi.service.*;
import com.itc.common.core.exception.CheckedException;
import com.itc.common.core.exception.ServiceException;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.Assert;

import java.sql.Timestamp;
import java.text.SimpleDateFormat;
import java.util.*;

@Service
@Slf4j
public class SqlCode extends DataSet {
    @Autowired
    DirectDataSetService dataSetService;
    @Autowired
    DatasourceService datasourceService;
    @Autowired
    DataSetParameterService dataSetParameterService;
    @Autowired
    private DirectoryService directoryService;
    @Autowired
    DataSetColumnService dataSetColumnService;
    @Autowired
    KafkaTemplate<String,Object> kafkaTemplate;
    @Autowired
    DataSynrecordMapper dataSynrecordMapper;
    @Autowired
    XxlJobInfoService xxlJobInfoService;

    @Transactional(rollbackFor = Exception.class)
    public void create(DataSetCreateDto dto){
        Assert.hasText(dto.getType(),"数据集类型不能为空");
        Assert.notNull(dto.getDtId(),"目录不能为空");
        Assert.notNull(dto.getName(),"数据集名称不能为空");
        Assert.notNull(dto.getCode(),"脚本不能为空");

        boolean editFlag = false;
        DirectDataSetEntity entity = null;
        if(null != dto.getId()){
            entity = dataSetService.getById(dto.getId());
            if(null == entity){
                throw new ServiceException("记录不存在");
            }
            entity.setDsId(dto.getDsId());
            entity.setCodeScript(dto.getCode());
            entity.setType(dto.getType());
            editFlag = true;
        }else{
            DirectoryEntity directory = directoryService.getById(dto.getDtId());
            if(null == directory || directory.getIsDir().equals(1)){
                throw new ServiceException("目录不存在");
            }
            entity = new DirectDataSetEntity();
            entity.setId(IdWorker.getId());

            entity.setDtId(dto.getDtId());
            entity.setDsId(dto.getDsId());
            entity.setCodeScript(dto.getCode());
            entity.setType(dto.getType());
        }

        DirectoryEntity directory = directoryService.getOne((Wrappers.<DirectoryEntity>lambdaQuery().eq(DirectoryEntity::getPid, dto.getDtId()).eq(DirectoryEntity::getName, dto.getName())));
        if(null != directory && !directory.getDtId().equals(dto.getId())) {
            throw new CheckedException("数据集名称已存在");
        }

        Long id = entity.getId();

        //参数处理
        if(dto.getParams().size() > 0){
            for (DataSetParameterEntity item : dto.getParams()){
                item.setDdsId(id);
            }
            dataSetParameterService.remove(Wrappers.lambdaQuery(DataSetParameterEntity.class).eq(DataSetParameterEntity::getDdsId,id));
            dataSetParameterService.saveBatch(dto.getParams());
        }
        DirectoryEntity dir = new DirectoryEntity();
        dir.setName(dto.getName());
        dir.setDtId(entity.getId());
        dir.setPid(dto.getDtId());
        dir.setIsDir(DataDirectorConstant.ISDATASET);

        List<DataSetColumnEntity> cloumnList = column(entity);


        if(editFlag){
            dataSetColumnService.validColumn(id,cloumnList);
            dataSetColumnService.remove(Wrappers.lambdaQuery(DataSetColumnEntity.class).eq(DataSetColumnEntity::getDataSetId,id));
            directoryService.updateById(dir);
        }else{
            directoryService.save(dir);
        }

        dataSetService.saveOrUpdate(entity);
        dataSetColumnService.saveBatch(cloumnList);

        if(editFlag){
            dataSetColumnService.redisCache(id);
            JSONObject data = new JSONObject();
            data.put("dtId",id);
            data.put("type","edit");
            kafkaTemplate.send(KafkaConstant.KAFKA_TOPIC_DATA_SET_COLUMN_EDIT,data.toJSONString());
            new Thread(new Runnable() {
                @Override
                public void run() {
                    job(xxlJobInfoService,dataSynrecordMapper,dto,"SqlHandler");
                }
            }).start();
        }else{
            //kafka推送消息
            JSONObject data = new JSONObject();
            data.put("dtId",entity.getId());
            data.put("cron",dto.getCron());
            kafkaTemplate.send(KafkaConstant.KAFKA_TOPIC_DATA_SET_SQL,data.toJSONString());
        }
    }

    @Override
    public List<Object> previewData(DataSetCreateDto dto){
        Assert.hasText(dto.getType(),"数据集类型不能为空");
        Assert.hasText(dto.getCode(),"脚本不能为空");
        if(dto.getCode().indexOf("${")>0 && dto.getParams().size()==0){
            throw new CheckedException("检测到脚本存在参数，但未配置参数，请检查");
        }
        Assert.notNull(dto.getDsId(),"数据源不能为空");
        //获取数据源
        DatasourceEntity datasource = getDadasource(dto.getDsId());
        //处理参数
        JSONObject param = getParam(dto.getParams());
        DatasourceHandler handler = DatasourceHandler.datasourceFactory(datasource);
        IPage page = handler.getPageDataBySql(dto.getCode(),1, SystemConstant.DATA_COLLECT_PREVIEW_SIZE,param);
        //日期格式化
        if(page.getRecords().size()>0){
            JSONObject data0 = (JSONObject)page.getRecords().get(0);
            Set<String> nullField = new HashSet<>();
            nullField.addAll(data0.keySet());
            Iterator<String> it=null;
            for (Object item : page.getRecords()){
                if(nullField.size()<1){
                    break;
                }
                it = nullField.iterator();
                JSONObject data = (JSONObject)item;
                while (it.hasNext()){
                    String filedName = it.next();
                    Object value = data.get(filedName);
                    if(null != value){
                        if(value instanceof java.sql.Date){
                            data.put(filedName, DateUtil.format((java.sql.Date) value, DatePattern.NORM_DATE_PATTERN));
                        }else if(value instanceof java.sql.Timestamp){
                            data.put(filedName,DateUtil.format((java.sql.Timestamp) value, DatePattern.NORM_DATETIME_PATTERN));
                        }else{
                            it.remove();
                        }
                    }
                }

            }
        }
        return page.getRecords();
    }

    /**
     * 分页参数为null时不分页
     * @param dataSet
     * @param pageIndex
     * @param pageSize
     * @return
     */
    @Override
    public List<Object> previewData(DirectDataSetEntity dataSet,Integer pageIndex,Integer pageSize){
        //获取数据源
        DatasourceEntity datasource = getDadasource(dataSet.getDsId());
        JSONObject param = getParam(dataSet.getId());
        DatasourceHandler handler = DatasourceHandler.datasourceFactory(datasource);
        List<Object> list = new ArrayList<>();
        if(null == pageIndex && null == pageSize){
            list.addAll(handler.getDataBySql(dataSet.getCodeScript(),param));
        }else{
            IPage page = handler.getPageDataBySql(dataSet.getCodeScript(),pageIndex,pageSize,param);
            list.addAll(page.getRecords());
        }
        handler.closeConn();
        return list;
    }

    @Override
    public List<Object> list(DirectDataSetEntity dataSet){
        //获取数据源
        DatasourceEntity datasource = getDadasource(dataSet.getDsId());
        JSONObject param = getParam(dataSet.getId());
        DatasourceHandler handler = DatasourceHandler.datasourceFactory(datasource);
        JSONArray arr = handler.getDataBySql(dataSet.getCodeScript(),param);
        handler.closeConn();
        return arr;
    }

    @Override
    public IPage page(DirectDataSetEntity dataSet,Integer pageIndex,Integer pageSize){
        //获取数据源
        DatasourceEntity datasource = getDadasource(dataSet.getDsId());
        JSONObject param = getParam(dataSet.getId());
        DatasourceHandler handler = DatasourceHandler.datasourceFactory(datasource);
        IPage page = handler.getPageDataBySql(dataSet.getCodeScript(),pageIndex,pageSize,param);
        handler.closeConn();
        return page;
    }

    @Override
    public List<DataSetColumnEntity> column(DirectDataSetEntity dataSet){
        DatasourceEntity datasource = getDadasource(dataSet.getDsId());
        //处理参数
        JSONObject param = getParam(dataSet.getId());
        String sql = "select t.* from ("+dataSet.getCodeScript()+" ) t where 1=2";
        DatasourceHandler handler = DatasourceHandler.datasourceFactory(datasource);
        List<DataSetColumnEntity> list = handler.getSqlColumn(sql,param);
        handler.closeConn();
        list.stream().forEach(item -> {
            /*if(item.getColumnName().matches(SystemConstant.REGEX_CHINESE)){
                throw new ServiceException("列名称不能包含汉字");
            }*/
            item.setDataSetId(dataSet.getId());
        });
        return list;
    }

    private DatasourceEntity getDadasource(Long id){
        DatasourceEntity datasource = datasourceService.getDecodeData(id);
        if(null == datasource){
            throw new ServiceException("数据源不存在");
        }
        return datasource;
    }

    //脚本参数设置
    private JSONObject getParam(Long dataSetId){
        //获取参数
        LambdaQueryWrapper<DataSetParameterEntity> qw = Wrappers.lambdaQuery();
        qw.eq(DataSetParameterEntity::getDdsId,dataSetId);
        List<DataSetParameterEntity> list = dataSetParameterService.list(qw);
        return getParam(list);
    }
    //脚本参数设置
    private JSONObject getParam(List<DataSetParameterEntity> list){
        //处理参数
        JSONObject param = new JSONObject();
        if(null != list && list.size()>0){
            list.stream().forEach(item -> {
                String name = item.getName();
                String type = item.getType();
                String value = item.getDefaultValue();

                if(SystemConstant.DATA_TYPE_STRING.equals(type)){
                    param.put(name,value);
                }else if(SystemConstant.DATA_TYPE_NUMBER.equals(type)){
                    if(value.indexOf(".")>0){
                        param.put(name,Double.valueOf(value));
                    }else{
                        param.put(name,Long.valueOf(value));
                    }
                }else if(SystemConstant.DATA_TYPE_DATE.equals(type)){
                    try {
                        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
                        param.put(name,sdf.parse(value));
                    }catch (Exception e){
                        throw new CheckedException("日期格式异常，请检查");
                    }
                }
            });
        }
        return param;
    }
}
